package com.kafka.kafka.study;

import com.kafka.kafka.utils.PropertiesUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * Project: tobe-better-coder
 * Package: com.kafka.kafka.study
 *
 * @author : Mr.xu
 * @date : 2023/9/10 15:19
 */
public class CustomProducerCallBack {

    public static void main(String[] args) {
        Properties properties = PropertiesUtils.getKafkaProperties("node1:9092,node2:9092");

        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.kafka.kafka.study.MyPartitioner");

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        for (int i = 0; i < 10; i++) {
            String msg;
            if(i%2 == 0){
               msg = "xuxin"+i;
            }else {
                msg = "kafka"+i;
            }
            kafkaProducer.send(new ProducerRecord<>("xuxin",1,"", "callBack111"), (recordMetadata, e) -> {
                if (null == e) {
                    System.out.println("topic = " + recordMetadata.topic() + " 分区 = " + recordMetadata.partition() +"msg = "+msg);
                }
            });
        }
        System.out.println(111);
        //没有这个消息发不出去
        kafkaProducer.close();

    }
}